package com.toutiao.vssue

import scala.collection.mutable.ArrayBuffer
import scala.collection.mutable.{HashMap => HMap}

import java.util.Date

import org.apache.log4j.Logger

import com.toutiao.AppProp

import com.toutiao.score.{Topic,Scorer,Measure}

import com.toutiao.model.{TimeDB,TimeDict,TimeIndex,Converter}

import com.toutiao.issue.SimpleDate

import net.liftweb.json.{JsonAST,Extraction,Printer,parse}
import net.liftweb.json.JsonDSL._


case class RawDocTitle (
            val id:Long,
            val create_time:Double,
            val cut_title:String,
            val images:Array[String]
)



class VssTitleModel(
        scorer: Scorer,
        consumer1: ConsumerStore,producer1: ProducerStore,
        consumer2: ConsumerStore,producer2: ProducerStore,
        minV:Double,minS:Double,
        locationDate:String,ID_URI: String,N:Int, //天
        DB_ROOT:String,DB_CACHE:Int,
        DOCID_NAME:String,SIMID_NAME:String
    ) {
  
  implicit val _ = net.liftweb.json.DefaultFormats
  
  private val log = Logger.getLogger(this.getClass.getName)
  
  var current = SimpleDate.current  
  
  val mDB = new TimeDB(DB_ROOT,DB_CACHE)
  val mTime = new TimeDict
  val K = 12
  val mIndex = new TimeIndex(K)
  val mID = new IDGetter(ID_URI)
  
  val mData = new VssTitleCollector(minV,mIndex,mTime,mDB,mID)
  log.info("Time Day Aware Model - Title")
  
  
  def parseLine(line:String) = {
    val u = new RawDocTitle(0L,0.0,"",Array(""))
    val s = 
        try {
            parse(line).extract[RawDocTitle]
        }
        catch {
            case _: Throwable => u
        }
    
    val i = s.id.toString
    val title = s.cut_title.split(" ")
    val image = 
      if (s.images.size < 3)
        s.images
      else
        s.images.slice(0, 3)
    val vs = scorer.score(title)
    val t = SimpleDate.parse((s.create_time*1000).toLong)
    
    new Document(i,title,image,vs,t)
  }
  
  def processOne(m:Document) = {
    val r = mData.process(m,minS)
    val i = m.name
    //log.info(i+" "+r)
    (i,r)
  }
  
  def process(ms:Array[Document]) = {
    log.info(ms.size)
    val ns = ms.map(e => processOne(e))
    log.info(ns.size)
    ns
  }

  def convOne(i:String,j:String) = {
    val m = Map[String,Long](DOCID_NAME->i.toLong,
                             SIMID_NAME->j.toLong)
    
    val z = Extraction.decompose(m)
    val line = Printer.compact(JsonAST.render(z))
    
    //i+" "+j
    line
  }
  def conv(ns:Array[(String,String)]) = {
    val us = ns.map(n => convOne(n._1,n._2))
    us
    //Event.send(us)
  }
  
  // 删除过期数据
  def remove(T:Date) = {
    val tt = SimpleDate.current
    val tt_s = SimpleDate.format(tt).split(" ")(0)
    val cu_s = SimpleDate.format(current).split(" ")(0)
    if (tt_s != cu_s) {
      current = tt
      val Ts = mDB.keySet.map(t => t + " 00:00:00")
        .map(t => SimpleDate.parse(t))
        .filter(_.getTime < T.getTime)
        .map(t => SimpleDate.format(t))
      if (Ts.size > 0) {
        log.info(Ts.toBuffer)
        Ts.foreach(t => {
          mData.remove(t)
        })
      }
    }
  }
  
  // 更新倒排与时间信息
  private def _init(xs:Array[(String,Topic)]) = {
    for (xx <- xs) {
      val x = xx._2
      val i = xx._1
      val t = SimpleDate.format(x.c).split(" ")(0)
      //log.info(t)
      mTime.insert(i,t)
      mIndex.insert(t,i.toLong,x.m("t"))
    }
  }
  
  private
  def _test(t:String,us:Array[String]) = {
    for (u <- us) {
      val xx = mDB(t).getValue(u)
      val zz = Converter.parseData(xx)
      log.info(u+" "+zz.i+" "+zz.m("t").toBuffer)
    }
  }
    
  // 初始化，Load Model
  def init = { 
    current = SimpleDate.loadDate(locationDate)
    val T1 = SimpleDate.increment(current,-N*60)
    val T2 = SimpleDate.increment(current, 60)
    
    var ct = T1
    
    while (ct.getTime < T2.getTime) {
      val t = SimpleDate.format(ct).split(" ")(0)
      log.info(t)
      mDB.init(t)
      val xs = mDB(t).getKVs
      log.info("Got "+xs.size)

      _init(xs)
      log.info("Index Done")
      
      ct = SimpleDate.increment(ct,60*24)
    }
    
    log.info(mDB.mData.keySet.toBuffer)
  }
  
  def run(m:Int) = synchronized {
    var current = SimpleDate.loadDate(locationDate)
    
    while (1==1) {
      
      val cc = 2048
      
      var cn = 2048
      
      while (cn > 128) {
        val ys = consumer1.get(cc)
        val zs = ys.keySet.toArray // !!!
        val xs = ys.values.toArray
        val ms = xs.map(x => parseLine(x)).filter(_.name != "0")
      
        log.info("Process 1... "+ms.size)
        val ns = process(ms)
        log.info("Process 1 Done")
      
        cn = ns.size
        if (cn > 0){
          val us = conv(ns)
          //us.foreach { u => log.info(u) }
          producer1.send(us)
        }      
        consumer1.ack(zs)
      }
        
      val ys = consumer2.get(cc)
      val zs = ys.keySet.toArray // !!!
      val xs = ys.values.toArray
      val ms = xs.map(x => parseLine(x)).filter(_.name != "0")
      
      log.info("Process 2... "+ms.size)
      val ns = process(ms)
      log.info("Process 2 Done")  
      
      val pn = ns.size
      if (pn > 0){
        val us = conv(ns)
        producer2.send(us)
      }
      consumer2.ack(zs)
      
      val T = SimpleDate.increment(current,-60*N)
      remove(T)
      
      val now = SimpleDate.current
      SimpleDate.saveDate(now, locationDate)
      
      if (pn <= 8){
        Thread.sleep(m*1000)
      }
    }
  }
  
}

object VssTitleProcess {
  
  def run(prop:AppProp) {
    
    val consumer1 = new ConsumerStore(
                      prop("ZOOKEEPER"),
                      prop("CRAWL_TOPIC1"),
                      prop("GROUP_ID"),
                      prop("CRAWL_TOPIC1_ACK")
                    )
    val producer1 = new ProducerStore(
                      prop("BROKERS"),
                      prop("SIM_TOPIC1")
                    )
    
    val consumer2 = new ConsumerStore(
                      prop("ZOOKEEPER"),
                      prop("CRAWL_TOPIC2"),
                      prop("GROUP_ID"),
                      prop("CRAWL_TOPIC2_ACK")
                    )
    val producer2 = new ProducerStore(
                      prop("BROKERS"),
                      prop("SIM_TOPIC2")
                    )
    
    //val locationAck1 = prop("CRAWL_TOPIC1_ACK")
    //val locationAck2 = prop("CRAWL_TOPIC1_ACK")
    
    val ID_URI = prop("ID_URI")
    
    val locationDate = prop("LOCATION_DATE")
    
    val N = prop("TIME_SIZE").toInt
    val minV = prop("MIN_V").toDouble
    val minS = prop("MIN_S").toDouble
    
    val dict = Measure.get(prop("LOCATION_MEASURE"))
    val n = prop("DOC_NUM").toInt
    val scorer = new Scorer(dict,n)    
 
    val DB_ROOT = prop("DB_ROOT")
    val DB_CACHE = prop("DB_CACHE").toInt
    
    val DOCID_NAME = prop("DOCID_NAME")
    val SIMID_NAME = prop("SIMID_NAME")    
    
    val m = new VssTitleModel(scorer,
                  consumer1,producer1,consumer2,producer2,
                  minV,minS,locationDate,ID_URI,
                  N,DB_ROOT,DB_CACHE,
                  DOCID_NAME,SIMID_NAME
                  )
    
    
    consumer1.run
    consumer2.run
    
    val K = prop("TIME_STEP").toInt
    m.init
    m.run(K)
  }
  
}


